/*
前提条件：添加JDBC用户机器IP到数据库白名单里，在pg_hba.conf添加以下内容即可：
假设JDBC用户IP为10.10.10.10
host    all             all             10.10.10.10/32        sha256
host    replication     all             10.10.10.10/32        sha256


创建复制槽条件
enable_slot_log = on
wal_level >= logical

用户有replication权限

*/

package cn.vastbase.jdbc.replication;




import cn.com.vastbase.PGProperty;
import cn.com.vastbase.jdbc.PgConnection;
import cn.com.vastbase.replication.LogSequenceNumber;
import cn.com.vastbase.replication.PGReplicationStream;

import java.nio.ByteBuffer;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class LogicalReplicationTestCase {
    private static PgConnection conn = null;

    public static void main(String[] args) throws SQLException {
        String driver = "cn.com.vastbase.Driver";
        //此处配置数据库IP以及端口，这里的端口为haPort，通常默认是所连接DN的port+1端口
        String sourceURL = "jdbc:vastbase://172.16.55.175:5000/test";
        //默认逻辑复制槽的名称是：replication_slot
        String slotName = "replication_slot";
        //测试模式：创建逻辑复制槽
        int TEST_MODE_CREATE_SLOT = 1;
        //测试模式：开启逻辑复制（前提条件是逻辑复制槽已经存在）
        int TEST_MODE_START_REPL = 2;
        //测试模式：删除逻辑复制槽
        int TEST_MODE_DROP_SLOT = 3;
        //开启不同的测试模式
        int testMode = 2;

        try {
            Class.forName(driver);
        } catch (Exception e) {
            e.printStackTrace();
            return;
        }

        try {
            Properties properties = new Properties();
            PGProperty.USER.set(properties, "zxs");
            PGProperty.PASSWORD.set(properties, "Vastdata#0730");
            //对于逻辑复制，以下三个属性是必须配置项
            PGProperty.ASSUME_MIN_SERVER_VERSION.set(properties, "9.4");
            PGProperty.REPLICATION.set(properties, "database");
            PGProperty.PREFER_QUERY_MODE.set(properties, "simple");
            conn = (PgConnection) DriverManager.getConnection(sourceURL, properties);
            System.out.println("connection success!");

            // 创建后查询 SELECT * FROM pg_replication_slots;
            if (testMode == TEST_MODE_CREATE_SLOT) {
                // 数据库接收命令： received wal replication command: CREATE_REPLICATION_SLOT replication_slot LOGICAL mppdb_decoding
                conn.getReplicationAPI()
                        .createReplicationSlot()
                        .logical()
                        .withSlotName(slotName) //这里字符串如包含大写字母则会自动转化为小写字母
//                        .withOutputPlugin("decoderbufs")
                        .withOutputPlugin("mppdb_decoding")
                        .make();

            } else if (testMode == TEST_MODE_START_REPL) {
                // 逻辑复制需要创建发布：CREATE PUBLICATION pub1 FOR ALL TABLES with(publish='insert,update,delete,truncate',ddl='all');
                //开启此模式前需要创建复制槽
                LogSequenceNumber waitLSN = LogSequenceNumber.valueOf("8/6B66ED0");

                PGReplicationStream stream = conn
                        .getReplicationAPI()
                        .replicationStream()
                        .logical()
                        .withSlotName(slotName)
//                        .withSlotOption("include-xids", false)
//                        .withSlotOption("skip-empty-xacts", true)
//                        .withStartPosition(waitLSN)
//                        .withSlotOption("parallel-decode-num", 1) //解码线程并行度
//                        .withSlotOption("white-table-list", "public.t1,public.t2") //白名单列表
//                        .withSlotOption("standby-connection", false) //强制备机解码
//                        .withSlotOption("decode-style", "t") //解码格式
//                        .withSlotOption("sending-batch", 1) //批量发送解码结果
//                        .withSlotOption("max-txn-in-memory", 100) //单个解码事务落盘内存阈值为100MB
//                        .withSlotOption("max-reorderbuffer-in-memory", 50) //正在处理的解码事务落盘内存阈值为50GB
                        .start();

                while (true) {
                    ByteBuffer byteBuffer = stream.readPending();

                    if (byteBuffer == null) {
                        TimeUnit.MILLISECONDS.sleep(10L);
                        continue;
                    }
                    int offset = byteBuffer.arrayOffset();
                    byte[] source = byteBuffer.array();
                    int length = source.length - offset;
                    System.out.println(new String(source, offset, length));

                    //如果需要flush lsn，根据业务实际情况调用以下接口
//                    LogSequenceNumber lastRecv = stream.getLastReceiveLSN();
//                    stream.setFlushedLSN(lastRecv);
//                    stream.forceUpdateStatus();

                }
            } else if (testMode == TEST_MODE_DROP_SLOT) {
                //  数据库接收命令：received wal replication command: DROP_REPLICATION_SLOT replication_slot
                conn.getReplicationAPI()
                        .dropReplicationSlot(slotName);

            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            conn.close();
        }
    }
}
